package com.webchat.connect.messagequeue.consumer.service;


import com.webchat.common.constants.ConnectConstants;
import com.webchat.common.enums.RoleCodeEnum;
import com.webchat.common.enums.messagequeue.MessageQueueEnum;
import com.webchat.common.service.messagequeue.producer.MessageQueueProducer;
import com.webchat.common.util.JsonUtil;
import com.webchat.connect.service.AccountService;
import com.webchat.connect.websocket.handler.ChatWebSocketEndPointServletHandler;
import com.webchat.domain.vo.request.mess.ChatMessageRequestVO;
import com.webchat.domain.vo.response.UserBaseResponseInfoVO;
import com.webchat.domain.vo.response.mess.ChatMessageResponseVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.Date;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;


/**
 * 负责处理对话WS请求消息，来自广播
 * （不包含音视频、红包等，文字对话+红包+文件传输等）
 */
@Slf4j
@Service
public class ChatMessageConsumeService {

    @Autowired
    private AccountService accountService;

    @Autowired
    private MessageQueueProducer<ChatMessageRequestVO, Long> messageQueueProducer;

    public void handleChatMessage(String message) {
        this.handleChatMessage(null, message);
    }

    /**
     * 处理来自对话topic 广播的消息
     * @param message
     */
    public void handleChatMessage(String bizCode, String message) {
        log.info("处理对话消息 =====> {}", message);
        ChatMessageRequestVO chatMessage = JsonUtil.fromJson(message, ChatMessageRequestVO.class);
        if (chatMessage == null) {
            return;
        }
        // 获取对话消息接受人
        String receiverId = chatMessage.getReceiverId();
        UserBaseResponseInfoVO receiver = accountService.accountInfo(receiverId);
        if (receiver == null) {
            return;
        }
        // 最终要推送的用户
        Set<String> receivers = new HashSet<>();
        /**
         * 处理一对一、群聊
         */
        // 构建消息响应VO，用于推送客户端
        ChatMessageResponseVO messageResponse = new ChatMessageResponseVO();
        messageResponse.setMessage(chatMessage.getMessage());
        messageResponse.setSenderId(chatMessage.getSenderId());
        messageResponse.setReceiverId(chatMessage.getReceiverId());
        messageResponse.setTime(new Date().getTime());
        messageResponse.setReceiverId(receiverId);
        messageResponse.setReceiver(receiver);
        messageResponse.setType(chatMessage.getType());
        if (RoleCodeEnum.GROUP.getCode().equals(receiver.getRoleCode())) {
            /**
             * 群聊(消息代理人，角色反转，原消息接受人-群聊反转为消息发送人，实际消息发送人作为被代理发送人)
             */
            // 由群组代理消息发送
            messageResponse.setSenderId(receiverId);
            // 被代理消息发送人
            messageResponse.setProxySenderId(chatMessage.getSenderId());
            UserBaseResponseInfoVO sender = accountService.accountInfo(chatMessage.getSenderId());
            messageResponse.setProxySender(sender);
            Set<String> groupUserIds = accountService.getGroupUserIds(receiverId);
            receivers.addAll(groupUserIds);
        } else if (RoleCodeEnum.isUserRole(receiver.getRoleCode())) {
            /**
             * 用户账号间一对一对话
             */
            receivers.add(receiverId);
        } else {
            log.error("不支持的对话场景 =====> receiverId:{}", receiverId);
            return;
        }

        /**
         * 《实时场景》消息推送
         */
        if (StringUtils.isNotBlank(bizCode)) {
            this.doPushMessage2Client(bizCode, messageResponse, receivers);
        } else {
            this.doPushMessage2Client(messageResponse, receivers);
        }
        /**
         * 《离线场景》持久化消息队列，保存离线消息，同时会将数据同步到ES用于后续的RAG问答和消息搜索
         *
         *  Consumer在UGC服务：
         *  com.webchat.ugc.messaegqueue.consumer.redis.PersistentMessageRedisMQConsumer
         *  com.webchat.ugc.messaegqueue.consumer.rocketmq.PersistentMessageRocketMQConsumer
         */
        messageQueueProducer.send(MessageQueueEnum.QUEUE_PERSISTENT_MESSAGE, chatMessage);
    }


    private void doPushMessage2Client(String bizCode, ChatMessageResponseVO chatMessage, Set<String> receivers) {
        Map<String, WebSocketSession> wsSessionMap = ChatWebSocketEndPointServletHandler.getSessions(bizCode, receivers);
        if (MapUtils.isEmpty(wsSessionMap)) {
            return;
        }
        chatMessage.setTime(System.currentTimeMillis());
        for (String receiver : receivers) {
            if (receiver.equals(chatMessage.getProxySenderId())) {
                continue;
            }
            chatMessage.setReceiverId(receiver);
            String message = JsonUtil.toJsonString(chatMessage);
            WebSocketSession socketSession = wsSessionMap.get(receiver);
            if (socketSession == null || !socketSession.isOpen()) {
                // 当前接受人可能离线、或者当前接受人的ws 链接在其他节点（
                // 广播后由其他节点处理 - 分布式下webscoket session共享，相见视频讲解）
                continue;
            }
            try {
                socketSession.sendMessage(new TextMessage(message));
            } catch (IOException e) {
                log.error("聊天消息推送异常 =====> receiver:{}, message:{}", receiver, message, e);
                // 这里如果异常可以增加重试策略 TODO
                // 但是不能阻塞其他人的消息推送（在多人群聊场景）
            }
        }
    }

    /**
     * 推送消息到客户端
     *
     * @param chatMessage
     * @param receivers
     */
    private void doPushMessage2Client(ChatMessageResponseVO chatMessage, Set<String> receivers) {

        Set<String> bizCodes = ConnectConstants.ConnectBiz.getBizCode(ConnectConstants.BizEnum.CHAT);
        for (String bizCode : bizCodes) {
            this.doPushMessage2Client(bizCode, chatMessage, receivers);
        }
    }

}
